package com.bw.medical.realtime.common.base;


import com.bw.medical.realtime.common.util.FlinkSourceUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

//    初始化方法 流数据
public abstract class BaseApp {
//    初始化
    public void start(String TopicDB,String GroupId,int Parallelism,int FlinkPort){
//        设置hadoop的执行用户
        System.setProperty("ROOT_USER_NAME","root");
//        创建流环境
        Configuration configuration=new Configuration();
//        flink配置端口
        configuration.setInteger("rest.port",FlinkPort);
        StreamExecutionEnvironment env= StreamExecutionEnvironment.getExecutionEnvironment(configuration);
        env.setParallelism(Parallelism);
////        设置检查点
//        env.setStateBackend(new HashMapStateBackend());
////        设置检查点超时时间
//        env.enableCheckpointing(5000);
////        设置检查点模式
//        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
////        设置检查点存储地址
//        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop103:8020/gmall/stream/"+GroupId);
////       最大并行度为1
//        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
////          设置检查点间隔
//        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(5000);
////        设置
//        env.getCheckpointConfig().setCheckpointTimeout(10000);
////        设置
//        env.getCheckpointConfig().setExternalizedCheckpointCleanup(RETAIN_ON_CANCELLATION);
//        消费数据
        DataStreamSource<String> dataStreamSource=env.fromSource(FlinkSourceUtil.getKafkaSource(TopicDB,GroupId), WatermarkStrategy.noWatermarks(),"kafka source");
//        打印数据
//        dataStreamSource.print();
//        初始化流
        handle(env,dataStreamSource);
        try {
            env.execute();
        }catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
    public abstract void handle(StreamExecutionEnvironment env, DataStreamSource<String> dataStreamSource);
}
